package com.chukun.flink.table.api.sql;

import com.chukun.flink.table.bean.Order;
import com.chukun.flink.table.bean.RateBean;
import com.chukun.flink.table.source.PrepareData;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TemporalTableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author chukun
 * @version 1.0.0
 * @description 时态表基本操作
 *
 *   时态表（Temporal Table）是一张随时间变化的表 – 在 Flink 中称为动态表，时态表中的每条记录都关联了一个或多个时间段，所有的 Flink 表都是时态的（动态的）。
 *   时态表包含表的一个或多个有版本的表快照，时态表可以是一张跟踪所有变更记录的表（例如数据库表的 changelog，包含多个表快照），也可以是物化所有变更之后的表（例如数据库表，只有最新表快照）
 *
 *   版本: 时态表可以划分成一系列带版本的表快照集合，表快照中的版本代表了快照中所有记录的有效区间，
 *      有效区间的开始时间和结束时间可以通过用户指定，根据时态表是否可以追踪自身的历史版本与否，时态表可以分为 版本表 和 普通表。
 *
 *  版本表: 如果时态表中的记录可以追踪和并访问它的历史版本，这种表我们称之为版本表，来自数据库的 changelog 可以定义成版本表。
 *  普通表: 如果时态表中的记录仅仅可以追踪并和它的最新版本，这种表我们称之为普通表，来自数据库 或 HBase 的表可以定义成普通表。
 *
 *
 *  时态表表示存储了多个数据维度的时间版本，这样在 join操作的时候，不会影响之前的数据
 *
 * @createTime 2022年06月03日 10:27:00
 */
public class JoinWithTemporalTable {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 加载汇率源数据
        DataStream<RateBean> rateStream = env.fromCollection(PrepareData.getRateData());

        // 将汇率的源数据转为带水印的数据流
        SingleOutputStreamOperator<RateBean> timedRateDataStream = rateStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((ctx) -> new WatermarkGenerator<RateBean>() {
                    @Override
                    public void onEvent(RateBean rate, long l, WatermarkOutput watermarkOutput) {
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                        watermarkOutput.emitWatermark(new Watermark(System.currentTimeMillis()));
                    }
                }).withTimestampAssigner((rate, timestamp) -> rate.getTime().getTime())
        );

        // 数据流转为数据表
        Table rateTable = tableEnv.fromDataStream(timedRateDataStream, $("id"), $("currency"), $("NowTime").rowtime(), $("rate"));

        // 注册汇率数据表
        tableEnv.createTemporaryView("RateHistory", rateTable);


        /**
         * 时态表解决问题:
         *   时态表表示存储了多个数据维度的时间版本，这样在 join操作的时候，不会影响之前的数据.
         */
        //创建一个时态表函数，将NowTime字段定义为时间属性，并将currency字段定义为主键
        TemporalTableFunction rateTemporalFun = rateTable.createTemporalTableFunction($("NowTime"), $("currency"));
        // 在表环境中注册一个名为 Rates的时态表函数，此时态表会存储每个时间段的汇率，类似于版本快照
        tableEnv.createTemporaryFunction("Rates", rateTemporalFun);


        // 加载订单源数据流
        DataStream<Order> orderStream = env.fromCollection(PrepareData.getOrderData());

        // 将订单的源数据转换为带水印的数据流
        SingleOutputStreamOperator<Order> timedOrderDataStream = orderStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((ctx) -> new WatermarkGenerator<Order>() {
                    @Override
                    public void onEvent(Order order, long timestamp, WatermarkOutput watermarkOutput) {}
                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                        watermarkOutput.emitWatermark(new Watermark(System.currentTimeMillis()));
                    }
                }).withTimestampAssigner((order, timestamp) -> order.getTime().getTime())
        );

        // 数据流转为数据表
        Table orderTable = tableEnv.fromDataStream(timedOrderDataStream, $("id"), $("currency"), $("orderTime").rowtime(), $("amount"));

        // 注册订单数据表
        tableEnv.createTemporaryView("Orders", orderTable);


        // 执行订单与时态表函数的连接语句
        String sql = "SELECT o.id,r.id,o.amount * r.rate AS amount " +
                "FROM Orders AS o,LATERAL TABLE (Rates(o.orderTime)) AS r " +
                "WHERE r.currency = o.currency";

        Table result = tableEnv.sqlQuery(sql);

        DataStream<Row> rowDataStream = tableEnv.toChangelogStream(result);

        rowDataStream.print("时态表");

        env.execute("JoinWithTemporalTable");

    }
}
